package com.sh.data.engine.domain.datadev.flinkTaskAboutExexte;

import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.FlinkUtil;
import com.sh.data.engine.common.util.FlinkUtil.FlinkCheckpointInfo;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkJobDomain;
import com.sh.data.engine.domain.datadev.flink.model.param.StartFlinkTaskParam;
import com.sh.data.engine.domain.datadev.flink.service.FlinkJobService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import com.sh.data.engine.domain.util.LogUtil;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkMenuNodeEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Objects;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/7/9 13:46
 */
@Service
@Slf4j
public class FlinkTaskApiService {

    @Autowired
    private FlinkSqlExcService flinkSqlExcService;
    @Autowired
    private FlinkAppExcService flinkAppExcService;
    @Autowired
    private FlinkNodeService flinkNodeService;
    @Autowired
    private FlinkJobService flinkJobService;

    public Long startFlinkTask(StartFlinkTaskParam param) {

        final FlinkMenuNodeEntity flinkMenuNode = flinkNodeService.getById(param.getNodeId());
        final Integer nodeType = flinkMenuNode.getNodeType();

        if (Objects.equals(nodeType, FlinkNodeTypeEnum.TASK_SQL.getCode())) {
            return flinkSqlExcService.clientJobStart(param);
        }

        if (Objects.equals(nodeType, FlinkNodeTypeEnum.TASK_JAR.getCode())) {
            return flinkAppExcService.clientJobStart(param);
        }

        throw new BusinessException(String.format("不支持的类型 %d", nodeType));
    }

    public void stopFlinkTask(Long jobId, Integer jobStatus, Boolean needSavepoint) {

        if (jobId == null) {
            throw new IllegalArgumentException("jobId can't be null");
        }

        log.info(
            "stop job ,jobId : {} ,jobStatus: {},needSavepoint: {},stackTrace: {}",
            jobId,
            jobStatus,
            needSavepoint,
            ExceptionUtils.getStackTrace(new Throwable()));

        flinkJobService.updateJobStatus(jobId, jobStatus);

        final FlinkJobDomain flinkJobDomain = flinkJobService.getByFlinkJobId(jobId);

        final Integer taskType = flinkJobDomain.getTaskType();

        boolean flink_yarn_mode = false;

        // flink app || flink sql
        if (taskType == FlinkNodeTypeEnum.TASK_SQL.getCode()
            || taskType == FlinkNodeTypeEnum.TASK_JAR.getCode()) {
            flink_yarn_mode = true;
        }

        // 目前的话FLINK_DATA_SYNC_TASK也会使用flink yarn 也会涉及到yarnApplicationIdsStr flinkJobIdsStr
        if (flink_yarn_mode) {
            // final String yarnApplicationIdsStr = flinkJobDomain.getYarnApplicationId();
            final String flinkJobId = flinkJobDomain.getFlinkJobId();

            if (StringUtils.isBlank(flinkJobId)) {
                return;
            }

            try {
                FlinkUtil.stopJob(String.valueOf(jobId), flinkJobId, needSavepoint);
            } catch (Exception e) {

            }
        }
    }

    public List<String> getFlinkTaskLogList(Long jobId) {
        if (jobId == null) {
            throw new IllegalArgumentException("jobId can't be null");
        }

        List<String> logList = Lists.newArrayList();

        final FlinkJobDomain jobDomain = flinkJobService.getByFlinkJobId(jobId);

        String logFileName = String.format("%s_%s", jobDomain.getNodeId(), jobDomain.getExecType());
        try {
            final List<String> strings = LogUtil.readFlinkLog(jobDomain.getNodeId(), 1);
            if (CollectionUtils.isNotEmpty(strings)) {
                logList.addAll(strings);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }

        return logList;
    }

    public FlinkCheckpointInfo getLastSavepoint(Long jobId) {
        final FlinkJobDomain flinkJobDomain = flinkJobService.getByFlinkJobId(jobId);
        final Integer taskType = flinkJobDomain.getTaskType();

        // 目前实时的采集中读取cdc数据源使用的也是flink sql模式
        if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_SQL.getCode())
            || Objects.equals(taskType, FlinkNodeTypeEnum.TASK_JAR.getCode())) {

            final String flinkJobId = flinkJobDomain.getFlinkJobId();

            if (StringUtils.isNotBlank(flinkJobId)) {
                try {
                    final List<FlinkCheckpointInfo> savePointList =
                        FlinkUtil.getSavePointList(String.valueOf(jobId), flinkJobId);
                    if (CollectionUtils.isNotEmpty(savePointList)) {
                        return savePointList.get(0);
                    }
                } catch (Exception ignore) {

                }
            }
        }
        return null;
    }

    public List<FlinkCheckpointInfo> getCheckpointList(Long jobId) {
        final FlinkJobDomain flinkJobDomain = flinkJobService.getByFlinkJobId(jobId);
        final Integer taskType = flinkJobDomain.getTaskType();

        if (Objects.equals(taskType, FlinkNodeTypeEnum.TASK_SQL.getCode())
            || Objects.equals(taskType, FlinkNodeTypeEnum.TASK_JAR.getCode())) {

            final String flinkJobId = flinkJobDomain.getFlinkJobId();

            if (StringUtils.isNotBlank(flinkJobId)) {
                try {
                    final List<FlinkCheckpointInfo> checkPointList =
                        FlinkUtil.getCheckPointList(String.valueOf(jobId), flinkJobId);
                    return checkPointList;
                } catch (Exception ignore) {

                }
            }
        }

        return Collections.EMPTY_LIST;
    }
}
